package kafka

import (
	"time"

	"github.com/IBM/sarama"
	"github.com/google/uuid"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
)

// KafkaCore 是 zapcore.Core 的实现，用于将日志写入 Kafka
type KafkaCore struct {
	encoder  zapcore.Encoder
	level    zapcore.LevelEnabler
	producer sarama.AsyncProducer
	topic    string
}

func NewKafkaCore(topic string, producer sarama.AsyncProducer, enc zapcore.Encoder, lvl zapcore.LevelEnabler) zapcore.Core {
	return &KafkaCore{
		topic:    topic,
		producer: producer,
		encoder:  enc,
		level:    lvl,
	}
}

func (c *KafkaCore) Enabled(level zapcore.Level) bool {
	return c.level.Enabled(level)
}

func (c *KafkaCore) With(fields []zapcore.Field) zapcore.Core {
	cloned := *c
	cloned.encoder = c.encoder.Clone()
	for _, f := range fields {
		f.AddTo(cloned.encoder)
	}
	return &cloned
}

func (c *KafkaCore) Check(entry zapcore.Entry, checkedEntry *zapcore.CheckedEntry) *zapcore.CheckedEntry {
	if c.Enabled(entry.Level) {
		return checkedEntry.AddCore(entry, c)
	}
	return checkedEntry
}

func (c *KafkaCore) Write(entry zapcore.Entry, fields []zapcore.Field) error {
	buf, err := c.encoder.EncodeEntry(entry, fields)
	if err != nil {
		return err
	}
	msg := &sarama.ProducerMessage{
		Topic: c.topic,
		Value: sarama.ByteEncoder(buf.Bytes()),
	}
	c.producer.Input() <- msg
	return nil
}

func (c *KafkaCore) Sync() error {
	return nil
}

// KafkaLoggerOption 配置项
type KafkaLoggerOption struct {
	Brokers  []string
	Topic    string
	Source   string
	Level    zapcore.Level // 如 zapcore.InfoLevel
	Username string
	Password string
}

// NewKafkaLogger 创建 zap.Logger 并输出日志到 Kafka
func NewKafkaLogger(opt KafkaLoggerOption) (*zap.Logger, error) {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = false
	config.Producer.Return.Errors = true
	config.Producer.Timeout = 3 * time.Second

	// 👇 加上这部分：设置SASL认证
	config.Net.SASL.Enable = true
	config.Net.SASL.User = opt.Username                  // 从 opt 里拿用户名
	config.Net.SASL.Password = opt.Password              // 从 opt 里拿密码
	config.Net.SASL.Mechanism = sarama.SASLTypePlaintext // 如果是PLAIN机制（一般Kafka默认就PLAIN）

	// 如果 Kafka 走的是 SASL_PLAINTEXT 协议，还要加这行（如果是SASL_SSL，就不用）
	config.Net.SASL.Handshake = true
	config.Net.TLS.Enable = false

	producer, err := sarama.NewAsyncProducer(opt.Brokers, config)
	if err != nil {
		return nil, err
	}

	encoderCfg := zapcore.EncoderConfig{
		TimeKey:        "time",
		LevelKey:       "level",
		NameKey:        "logger", // LoggerName
		CallerKey:      "caller", // 调用者文件路径
		FunctionKey:    "func",   // 函数名（需开启 EnableFunctionName）
		MessageKey:     "data",
		StacktraceKey:  "stacktrace",
		EncodeTime:     zapcore.ISO8601TimeEncoder,
		EncodeLevel:    zapcore.LowercaseLevelEncoder,
		EncodeCaller:   zapcore.ShortCallerEncoder,
		EncodeDuration: zapcore.SecondsDurationEncoder,
	}

	encoder := zapcore.NewJSONEncoder(encoderCfg)

	core := NewKafkaCore(opt.Topic, producer, encoder, opt.Level)
	logger := zap.New(core).With(
		zap.String("id", uuid.NewString()),
		zap.String("source", opt.Source),
		zap.String("type", "log.v1"),
	)

	return logger, nil
}
